package com.zlzlib.libnetty.ws;

import com.zlzlib.libnetty.base.BaseClient;
import com.zlzlib.libnetty.listener.WebSocketListener;
import com.zlzlib.libnetty.util.Util;
import com.zlzlib.libnetty.ws.handler.HeartbeatWebHandler;

import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * webSocket客户端
 *
 * @DateTime: 2021/4/23 10:30
 * @Author zlz
 * @Version 1.0
 */
public class WebSocketClient extends BaseClient<Object> implements WebSocketListener {

    public WebSocketClient() {
        headers = new DefaultHttpHeaders();
    }

    private WebChannelInitializer initializer;
    private final HttpHeaders headers;
    private CountDownLatch mCountDownLatch;

    @Override
    protected void initChannel() {
    }

    @Override
    protected void initChannel(URI url) {
        Util.logInfo("headers: " + headers);
        if (mEventListener != null) {
            initializer = mEventListener.getWebChannel(this, url, headers);
        }
        if (bootstrap == null) {
            return;
        }
        if (initializer != null) {
            bootstrap.handler(initializer);
        } else {
            bootstrap.handler(initializer = new WebChannelInitializer(this, url, headers));
        }
    }

    @Override
    protected void toServer() {
        try {
            channel = (SocketChannel) bootstrap.connect(currentHost, currentPort).sync().channel();
        } catch (Exception e) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.err.printf("连接Server(ip[%s], port[%s])失败%n", currentHost, currentPort);
            channel = null;
        }
    }

    @Override
    protected void startHandshake() {
        //握手
        try {
            initializer.handshakeFuture();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void addHeartbeatHandler() {
        if (channel == null || !channel.isActive() || channel.pipeline() == null) {
            return;
        }
        try {
            // 之前存在的读写超时handler，先移除掉，再重新添加
            if (channel.pipeline().get(IdleStateHandler.class.getSimpleName()) != null) {
                channel.pipeline().remove(IdleStateHandler.class.getSimpleName());
            }
            if (channel.pipeline().get(HeartbeatWebHandler.class.getSimpleName()) != null) {
                channel.pipeline().remove(HeartbeatWebHandler.class.getSimpleName());
            }
            // 3次心跳没响应，代表连接已断开
            channel.pipeline().addFirst(IdleStateHandler.class.getSimpleName(), new IdleStateHandler(
                    3L * getHeartbeatInterval(), getHeartbeatInterval(), 0, TimeUnit.MILLISECONDS));
            // 重新添加HeartbeatHandler
            channel.pipeline().addLast(HeartbeatWebHandler.class.getSimpleName(), new HeartbeatWebHandler(this));
        } catch (Exception e) {
            e.printStackTrace();
            Util.logError("添加心跳消息管理handler失败，reason：" + e.getMessage());
        }
    }

    @Override
    protected void sendSelfMsg(final Object msg) {
        try {
            WebSocketFrame webSocketFrame;
            if (msg instanceof byte[]) {
                byte[] bytes = (byte[]) msg;
                ByteBuf buffer = this.channel.alloc().buffer(bytes.length);
                webSocketFrame = new BinaryWebSocketFrame(buffer.writeBytes(bytes));
            } else {
                webSocketFrame = new TextWebSocketFrame(msg.toString());
            }
            ChannelFuture future = channel.writeAndFlush(webSocketFrame);
            future.addListener((ChannelFutureListener) channelFuture -> {
                Util.logInfo("发送消息，message=" + msg + "  " + channelFuture.isSuccess());
                mEventListener.msgSendState(msg, channelFuture.isSuccess());
            });
        } catch (Exception ex) {
            Util.logInfo("发送消息失败，reason:" + ex.getMessage() + "\tmessage=" + msg);
            mEventListener.msgSendState(msg, false);
        }
    }

    public synchronized long ping() {
        mCountDownLatch = new CountDownLatch(1);
        long timeMillis = System.currentTimeMillis();
        //发送ping消息，接收到后，返回延迟时间
        try {
            if (isAlive()) {
                this.channel.writeAndFlush(new PingWebSocketFrame());
                mCountDownLatch.await();
                return System.currentTimeMillis() - timeMillis;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 99999;
    }

    /**
     * 添加信息头  请在初始化之前调用
     * 或者调用之后重新连接   resetConnect(true);
     */
    public void addHeader(String name, Object value) {
        if (headers.contains(name)) {
            headers.remove(name);
        }
        headers.add(name, value);
    }

    @Override
    public void onOpen(FullHttpResponse response) {
        addHeartbeatHandler();
    }

    @Override
    public void onMessage(String text) {
        getMsgDispatcher().receivedMsg(text);
    }

    @Override
    public void onMessage(byte[] bytes) {
        getMsgDispatcher().receivedMsg(bytes);
    }

    @Override
    public void onClosed(int code, String reason) {
        //关闭
        if (mStatusListener != null) mStatusListener.onConnectClose();
    }

    @Override
    public void onFailure(Throwable t, FullHttpResponse response) {
        if (mStatusListener != null) mStatusListener.onConnectException(t);
        resetConnect();
    }

    @Override
    public void onHeartPong() {
        //这里需要计时
        if (mCountDownLatch != null) mCountDownLatch.countDown();
    }
}
